﻿#include "thread_pool.h"
#include "thread_mutex.h"
#include "common.h"

#include <string.h>
#include <signal.h>
#include <errno.h>
#include <sys/eventfd.h>

#include <sys/select.h>
#include <sys/time.h>
#include <sys/types.h>
#include <unistd.h>

#include <sched.h>

#define MIN_THREAD_NUM      3
#define MAX_THREAD_NUM      100

#define THREAD_WAIT_NUM_GAP      8
#define THREAD_MAX_SUB_NUM       3

static bool thread_is_run = false;

static thread_pool_queue_t  thread_pool_done;
static pthread_spinlock_t   spin_lock;
static unsigned int         thread_pool_task_id;
static int                  eventValue;

void* thread_pool_cycle(void* data)
{
    thread_pool_t* tp = (thread_pool_t*)data;

    int                 err;
    sigset_t            set;
    thread_task_t*      task;
    char errStr[1024] = { 0 };

    sigfillset(&set);

    sigdelset(&set, SIGILL);
    sigdelset(&set, SIGFPE);
    sigdelset(&set, SIGSEGV);
    sigdelset(&set, SIGBUS);

    err = pthread_sigmask(SIG_BLOCK, &set, NULL);
    if (0 != err)
    {
        LOG2(ERROR, 
            "pthread_sigmask error, info : %s", 
            strerror(errno)
        );

        return NULL;
    }

    for (;; ) 
    {
        if (0 != mutex_lock(errStr, &tp->m_mutex)) 
        {
            LOG2(ERROR,
                "mutex_lock error, info : %s",
                errStr
            );

            return NULL;
        }

        while (tp->m_queue.first == NULL) 
        {
            if (pthread_cond_wait(&tp->m_cond, &tp->m_mutex))
            {
                LOG2(ERROR, 
                    "pthread_cond_wait error , info : %s", 
                    strerror(errno)
                );

                mutex_lock(errStr, &tp->m_mutex);
                return NULL;
            }
        }

        /* the number may become negative */
        tp->m_waiting--;

        task = tp->m_queue.first;
        tp->m_queue.first = task->next;

        if (tp->m_queue.first == NULL) 
        {
            tp->m_queue.last = &tp->m_queue.first;
        }

        if (0 != mutex_unlock(errStr, &tp->m_mutex)) 
        {
            LOG2(ERROR, 
                "mutex_unlock error, info : %s", 
                strerror(errno)
            );

            return NULL;
        }

        task->handler(task->ctx);  /* 任务处理 */
        task->next = NULL;
        
        err = pthread_spin_lock(&spin_lock);
        if (0 != err)
        {
            LOG2(ERROR, 
                "Error on pthread_spin_lock, info : %s", 
                strerror(err)
            );
            return NULL;
        }

        *thread_pool_done.last = task;
        thread_pool_done.last = &task->next;


        pthread_spin_unlock(&spin_lock);

        thread_pool_handler();
    }
}

void* thread_pool_manage(void* data)
{
    thread_pool_t* tp = (thread_pool_t*)data;
    struct timeval tv = { 5, 0 };
    pthread_attr_t       attr;

    pthread_t tid;
    int ret = 0, sub_num, add_num;

    volatile unsigned int lock;
    char errStr[1024] = { 0 };

    fd_set fdSet;
    thread_task_t task;

    eventValue = eventfd(0, 0);
    if (eventValue < 0)
    {
        LOG2(ERROR, "eventfd error, %s", strerror(errno));
        return NULL;
    }

    ret = pthread_attr_init(&attr);
    if (0 != ret)
    {
        LOG2(ERROR,
            "Error on pthread_attr_init , info : %s",
            strerror(ret)
        );
        return NULL;
    }

    ret = pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED);
    if (0 != ret)
    {
        LOG2(ERROR,
            "Error on pthread_attr_setdetachstate , info : %s",
            strerror(ret)
        );

        return NULL;
    }

    while (thread_is_run)
    {
        tv.tv_sec = 5;
        tv.tv_usec = 0;

        FD_ZERO(&fdSet);
        FD_SET(eventValue, &fdSet);

        ret = select(eventValue + 1, &fdSet, NULL, NULL, &tv);
        if (0 > ret)
        {
            LOG2(ERROR, 
                "Error on select, info :%s", 
                strerror(errno)
            );

            return NULL;
        }

        if (FD_ISSET(eventValue, &fdSet))
        {
            LOG2(INFO, "management thread exit");
            break;/* 结束退出 */
        }

        if (tp->m_waiting > tp->m_threads && tp->m_threads < tp->m_max_threads)
        {
            add_num = tp->m_max_threads - tp->m_threads;
            if (add_num > tp->m_add_thread) add_num = tp->m_add_thread;

            tp->m_threads += add_num;
            for (int i = 0; i < add_num; ++i)
            {
                ret = pthread_create(&tid, &attr, thread_pool_cycle, (void*)tp);
                if (0 != ret)
                {
                    LOG2(ERROR,
                        "thread_pool_manage create thread failed, info : %s",
                        strerror(ret)
                    );

                    return NULL;
                }
            }

            LOG2(INFO,
                "thread pool add %d worker, cur worker num %d",
                add_num, tp->m_threads.load()
            );
        }
        else if ((tp->m_threads > MIN_THREAD_NUM) && (tp->m_waiting == 0))
        {
            sub_num = tp->m_threads - MIN_THREAD_NUM;
            if (sub_num > tp->m_sub_thread)
            {
                sub_num = tp->m_sub_thread;
            }
            tp->m_threads -= sub_num;

            task.handler = thread_exit_handler;
            task.ctx = (void*)&lock;

            for (int i = 0; i < sub_num; ++i)
            {
                lock = 1;
                task.ev.complete = 0;
                task.ev.active = 0;

                if (0 != thread_task_post(errStr, tp, &task))
                {
                    LOG2(ERROR, "thread pool delete idle threads fail, info : %s", errStr);

                    return NULL;
                }

                while (lock)
                {
                    sched_yield();
                }
            }

            LOG2(INFO, 
                "End idle thread, sub num : %d , cur num : %d", 
                sub_num, tp->m_threads.load()
            );
        }
    }
    
    pthread_attr_destroy(&attr);

    return NULL;
}

void thread_exit_handler(void* data)
{
    int* lock = (int*)data;

    *lock = 0;

    pthread_exit(0);
}

int thread_pool_init(char* errStr, thread_pool_t* tp)
{
    int                  err;
    pthread_t            tid;
    pthread_attr_t       attr;

    if ((tp->m_threads < MIN_THREAD_NUM) || (tp->m_threads > tp->m_max_queue) || (tp->m_threads > MAX_THREAD_NUM))
    {
        sprintf(errStr,
            "invalid parameter, min threadnum : %d, max thread  num : %d, thread num : %d",
            MIN_THREAD_NUM, MAX_THREAD_NUM, tp->m_threads.load()
        );

        return -1;
    }

    if (tp->m_max_threads < MIN_THREAD_NUM)
    {
        tp->m_max_threads = MIN_THREAD_NUM;
    }
    else if (tp->m_max_threads > MAX_THREAD_NUM)
    {
        tp->m_max_threads = MAX_THREAD_NUM;
    }

    thread_pool_task_id = 0;

    ngx_thread_pool_queue_init(&tp->m_queue);
    ngx_thread_pool_queue_init(&thread_pool_done);

    if (0 != mutex_create(errStr, &tp->m_mutex))
    {
        return -1;
    }

    err = pthread_cond_init(&tp->m_cond, NULL);
    if (0 != err)
    {
        sprintf(errStr,
            "Error on pthread_cond_init , info : %s",
            strerror(err)
        );

        return -1;
    }

    err = pthread_attr_init(&attr);
    if (0 != err)
    {
        sprintf(errStr,
            "Error on pthread_attr_init , info : %s",
            strerror(err)
        );
        return -1;
    }

    err = pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED);
    if (0 != err)
    {
        sprintf(errStr,
            "Error on pthread_attr_setdetachstate , info : %s",
            strerror(err)
        );

        return -1;
    }

    err = pthread_spin_init(&spin_lock, PTHREAD_PROCESS_PRIVATE);
    if (0 != err)
    {
        sprintf(errStr,
            "Error on pthread_spin_init , info : %s",
            strerror(err)
        );

        return -1;
    }

    thread_is_run = true;
    for (int n = 0; n < tp->m_threads; n++)
    {
        err = pthread_create(&tid, &attr, thread_pool_cycle, (void*)tp);
        if (0 != err)
        {
            LOG2(ERROR, "pthread_create() failed");
            return -1;
        }
    }

    err = pthread_create(&tid, &attr, thread_pool_manage, (void*)tp);
    if (0 != err)
    {
        LOG2(ERROR,
            "thread_pool_manage start failed, info %s",
            strerror(err)
        );
        return -1;
    }

    pthread_attr_destroy(&attr);

    return 0;
}

int thread_task_post(char* errStr, thread_pool_t* tp, thread_task_t* task)
{
    if (task->ev.active) {
        sprintf(errStr,
            "task : %u already active", 
            task->id
        );

        return -1;
    }

    if (0 != mutex_lock(errStr, &tp->m_mutex)) 
    {
        return -1;
    }

    if (tp->m_waiting >= tp->m_max_queue) 
    {
        mutex_unlock(errStr, &tp->m_mutex);

        LOG2(ERROR, 
            "thread pool queue overflow: %d tasks waiting", 
            tp->m_waiting
        );

        return -1;
    }

    task->ev.active = 1;
    task->id = thread_pool_task_id++;
    task->next = NULL;

    if (0 != pthread_cond_signal(&tp->m_cond))
    {
        mutex_unlock(errStr, &tp->m_mutex);
        
        LOG2(ERROR, 
            "pthread_cond_signal error , info : %s", 
            strerror(errno)
        );
    
        return -1;
    }

    *tp->m_queue.last = task;
    tp->m_queue.last = &task->next;

    tp->m_waiting++;

    mutex_unlock(errStr, &tp->m_mutex);
    LOG2(INFO, "new task added to thread pool");

    return 0;
}

int thread_pool_destroy(char*errStr, thread_pool_t* tp)
{
    uint64_t  evefd = 1;
    thread_task_t    task;
    volatile unsigned int lock;

    pthread_spin_destroy(&spin_lock);

    if (sizeof(uint64_t) != write(eventValue, &evefd, sizeof(uint64_t)))
    {
        LOG2(ERROR, "Error on shutdown : %s", strerror(errno));
        return -1;
    }

    memset(&task, 0x00, sizeof(thread_task_t));

    task.handler = thread_exit_handler;
    task.ctx = (void*)&lock;

    for (int n = 0; n < tp->m_threads; n++) 
    {
        lock = 1;
        task.ev.active = 0;
        task.ev.complete = 0;

        if (0 != thread_task_post(errStr, tp, &task)) 
        {
            return -1;
        }

        while (lock) 
        {
            sched_yield();
        }
    }
    mutex_destroy(errStr, &tp->m_mutex);

    return 0;
}

int thread_pool_handler()
{
    event_t* event;
    thread_task_t* task;

    int err = pthread_spin_lock(&spin_lock);
    if (0 != err)
    {
        LOG2(ERROR,
            "Error on pthread_spin_lock, info : %s",
            strerror(err)
        );
        return -1;
    }

    task = thread_pool_done.first;
    thread_pool_done.first = NULL;
    thread_pool_done.last = &thread_pool_done.first;

    pthread_spin_unlock(&spin_lock);

    while (task) {
        LOG2(INFO,
            "run completion handler for task_id : %u", 
            task->id
        );

        event = &task->ev;
        task = task->next;

        event->complete = 1;
        event->active = 0;

        if(event->handler) event->handler(event);
    }

    return 0;
}